/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.rocketmq.client.latency;

import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.logging.InternalLogger;

/**
 * Producer的负载均衡
 *
 * Producer端在发送消息的时候，会先根据Topic找到指定的TopicPublishInfo，在获取了TopicPublishInfo路由信息后，RocketMQ的客户端在默认方式下 {@link #selectOneMessageQueue}方法会从TopicPublishInfo中的messageQueueList中选择一个队列（MessageQueue）进行发送消息。
 * 这里有一个 {@link #sendLatencyFaultEnable}开关变量，如果开启，在随机递增取模的基础上，再过滤掉not available的Broker。
 * 所谓的"latencyFaultTolerance"，是指对之前失败的，按一定的时间做退避。例如，如果上次请求的latency超过550Lms，就退避3000Lms；超过1000L，就退避60000L；
 * 如果关闭，采用随机递增取模的方式选择一个队列（MessageQueue）来发送消息，latencyFaultTolerance机制是实现消息发送高可用的核心关键所在。
 */
public class MQFaultStrategy {
    private final static InternalLogger log = ClientLogger.getLog();
    private final LatencyFaultTolerance<String> latencyFaultTolerance = new LatencyFaultToleranceImpl();

    /**
     * 发送mq消息时,是否启用broker故障延迟机制
     *
     * {@link #selectOneMessageQueue} 选择一个消息队列
     *
     * <pre>
     *     首先在一次消息发送过程中,可能会多次执行选择消息队列这个方法, lastBrokerName就是上一次选择的执行发送消息失败的Broker
     *     第一次执行消息队列选择时,lastBrokerName为null,此时直接用sendWhichQueue自增再获取值,
     *     与当前路由表中消息队列个数取模,返回该位置的MessageQueue(selectOneMessageQueue()方法),
     *     如果消息发送再失败的话,下次进行消息队列选择时规避上次MesageQueue所在的Broker,否则还是很有可能再次失败
     *
     *     该算法在一次消息发送过程中能成功规避故障的Broker,但如果Broker启机,由于路由算法中的消息队列是按Broker排序的,
     *     如果上一次根据路由算法选择的是宕机的Broker的第一个队列,那么随后的下次选择的是启机Broker的第二个队列,消息发送很有可能会失败,
     *     再次引发重试,带来不必要的性能损耗,那么有什么方法在一次消息发送失败后,暂时将该Broker排除在消息队列选择范围外呢?
     *     或许有朋友会问, Broker不可用后,路由信息中为什么还会包含该Broker的路由信息呢?
     *     其实这不难解释:首先,NameServer检测Broker是否可用是有延迟的,最短为一次心跳检测间隔(10s);
     *     其次, NameServer不会检测到Broker启机后马上推送消息给消息生产者,而是消息生产者每隔30s更新一次路由信息,所以消息生产者最快感知Broker最新的路由信息也需要30s
     *     如果能引入一种机制,在Broker启机期间,如果一次消息发送失败后,可以将该Broker暂时排除在消息队列的选择范围中。
     * </pre>
     */
    private boolean sendLatencyFaultEnable = false;

    private long[] latencyMax =           {50L, 100L, 550L,   1000L,  2000L,   3000L,   15000L};
    private long[] notAvailableDuration = {0L,  0L,   30000L, 60000L, 120000L, 180000L, 600000L};

    public long[] getNotAvailableDuration() {
        return notAvailableDuration;
    }

    public void setNotAvailableDuration(final long[] notAvailableDuration) {
        this.notAvailableDuration = notAvailableDuration;
    }

    public long[] getLatencyMax() {
        return latencyMax;
    }

    public void setLatencyMax(final long[] latencyMax) {
        this.latencyMax = latencyMax;
    }

    public boolean isSendLatencyFaultEnable() {
        return sendLatencyFaultEnable;
    }

    public void setSendLatencyFaultEnable(final boolean sendLatencyFaultEnable) {
        this.sendLatencyFaultEnable = sendLatencyFaultEnable;
    }

    /**
     * 从topic中的多个queue中选择其中一个queue.
     *
     * @param tpInfo         topic信息
     * @param lastBrokerName 不返回broker名称的队列
     * @return 消息队列信息
     */
    public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        // producer发送mq消息时,负载均衡策略是否启用
        if (!this.sendLatencyFaultEnable) {
            // sendLatencyFaultEnable = false,未启用情况,直接让 TopicPublishInfo 选择一个queue
            return tpInfo.selectOneMessageQueue(lastBrokerName);
        }

        try {
            int index = tpInfo.getSendWhichQueue().incrementAndGet();
            for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                if (pos < 0)
                    pos = 0;
                MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
                    return mq;
            }

            final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
            int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
            if (writeQueueNums > 0) {
                final MessageQueue mq = tpInfo.selectOneMessageQueue();
                if (notBestBroker != null) {
                    mq.setBrokerName(notBestBroker);
                    mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
                }
                return mq;
            } else {
                latencyFaultTolerance.remove(notBestBroker);
            }
        } catch (Exception e) {
            log.error("Error occurred when selecting message queue", e);
        }

        return tpInfo.selectOneMessageQueue();
    }

    public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
        if (this.sendLatencyFaultEnable) {
            long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
            this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
        }
    }

    private long computeNotAvailableDuration(final long currentLatency) {
        for (int i = latencyMax.length - 1; i >= 0; i--) {
            if (currentLatency >= latencyMax[i])
                return this.notAvailableDuration[i];
        }

        return 0;
    }
}
